package com.swak.reactivex.cluster.impl;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.SocketTimeoutException;
import java.net.StandardSocketOptions;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.exception.RemotingException;
import com.swak.reactivex.cluster.Networker;
import com.swak.registry.URL;
import com.swak.spi.SPI;
import com.swak.timer.Crontab;
import com.swak.utils.Nets;
import com.swak.utils.Sets;

/**
 * MulticastNetworker.
 * 
 * @author 618lf
 */
@SPI(value = "Multicast")
public class MulticastNetworker implements Networker {
	private static final Logger LOGGER = LoggerFactory.getLogger(Networker.class);
	private static final String JOIN = "join";
	private static final String LEAVE = "leave";
	private final Thread loop;
	private volatile boolean running = true;
	private InetAddress address;
	private MulticastSocket socket;
	private Set<NetworkerListener> listeners = Sets.newConcurrentSet();
	private final Crontab crontab;
	private final Set<URL> services = Sets.newConcurrentSet();
	private int discoveryTimes;

	public MulticastNetworker(URL url) throws IOException {
		String multicastHost = url.getParameter(Constants.Discovery_Multicast_Host_Key,
				Constants.Discovery_Multicast_Host);
		int multicastPort = url.getParameter(Constants.Discovery_Multicast_Port_Key, Constants.Multicast_Port);
		this.discoveryTimes = url.getParameter(Constants.Discovery_Multicast_Discovery_Times_Key,
				Constants.Multicast_Discovery_Times);
		if (!Nets.isMulticastAddress(multicastHost)) {
			throw new IllegalArgumentException(
					"Invalid multicast address " + multicastHost + ", scope: 224.0.0.0 - 239.255.255.255");
		}
		this.address = InetAddress.getByName(multicastHost);
		this.socket = new MulticastSocket(new InetSocketAddress(multicastPort));// 使用 InetAddress.anyLocalAddress()
		this.socket.setOption(StandardSocketOptions.IP_MULTICAST_LOOP, true); // 开启此参数表示接收当前主机的信息，利于测试
		this.socket.setSoTimeout(url.getParameter(Constants.Timeout, 10000)); // receive timeout
		this.socket.joinGroup(new InetSocketAddress(this.address, multicastPort), Nets.getLocalNetwork());
		this.crontab = new Crontab("0 0/1 * * * *"); // 每分钟执行一次
		this.loop = new Thread(() -> {
			while (running) {
				this.receive();
				this.joinTask();
			}
		}, "Multicast-Discovery");
		this.loop.setDaemon(true);
		this.loop.start();

		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("Using Multicast Discovery");
		}
	}

	private void joinTask() {
		if (this.discoveryTimes > 0) {
			this.crontab.execute(() -> {
				this.discoveryTimes--;
				this.services.forEach((url) -> {
					this.join(url);
				});
			});
		}
	}

	private void receive() {
		try {
			byte[] buf = new byte[1024];
			DatagramPacket recv = new DatagramPacket(buf, buf.length);
			this.socket.receive(recv);
			this.received(new String(recv.getData()).trim());
		} catch (SocketTimeoutException e) {
		} catch (Exception e) {
			LOGGER.error(e.getMessage(), e);
		}
	}

	protected void received(String message) throws RemotingException {
		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("Multicast Recv:" + message);
		}
		for (NetworkerListener listener : listeners) {
			if (message.startsWith(JOIN)) {
				String url = message.substring(JOIN.length()).trim();
				listener.join(URL.valueOf(url));
			} else if (message.startsWith(LEAVE)) {
				String url = message.substring(LEAVE.length()).trim();
				listener.leave(URL.valueOf(url));
			}
		}
	}

	@Override
	public void listen(NetworkerListener listener) {
		listeners.add(listener);
	}

	private void send(String msg) {
		DatagramPacket hi = new DatagramPacket(msg.getBytes(), msg.length(), address, this.socket.getLocalPort());
		try {
			this.socket.send(hi);
		} catch (IOException e) {
			throw new IllegalStateException(e.getMessage(), e);
		}
	}

	@Override
	public void join(URL url) {
		services.add(url);
		String message = JOIN + " " + url.toFullString();
		send(message);
		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("Multicast Send:" + message);
		}
	}

	@Override
	public void leave(URL url) {
		services.remove(url);
		String message = LEAVE + " " + url.toFullString();
		send(message);
		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("Multicast Send:" + message);
		}
	}

	@Override
	public void close() {
		services.clear();
		this.running = false;
	}
}